package com.lagou.filter;

import com.google.gson.JsonObject;
import com.lagou.pojo.Monitor;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @ClassName: TPMonitorFilter
 * @Auther: Jerry
 * @Date: 2020/10/19 9:13
 * @Desctiption: 利用TPMonitorFilter在消费端记录每个方法的请求耗时时间
 * @Version: 1.0
 */
@Activate(group = {CommonConstants.CONSUMER, CommonConstants.PROVIDER})
public class TPMonitorFilter implements Filter, Runnable {

    // 耗时监控集合
    private ConcurrentMap<String, List<Monitor>> MONITOR_MAP = new ConcurrentHashMap<>();

    /**
     *
     * 每隔5s打印一次最近1分钟内每个方法TP90、TP99的耗时情况
     */
    public TPMonitorFilter(){
        // 每隔5s打印一次最近1分钟内每个方法TP90、TP99的耗时情况
        Executors.newSingleThreadScheduledExecutor()
                .scheduleWithFixedDelay(this, 1, 5, TimeUnit.SECONDS);

        //每60秒删除一次过期的数据
        Executors.newSingleThreadScheduledExecutor().scheduleWithFixedDelay(()->{
            long now = System.currentTimeMillis() - 60 * 1000;
            MONITOR_MAP.forEach((key,value)-> value.removeIf(item -> item.getEndTime() < now));
        }, 60, 60, TimeUnit.SECONDS);
    }

    @Override
    public void run() {
//        synchronized (MONITOR_MAP) {
            long now = System.currentTimeMillis() - 60 * 1000;
            MONITOR_MAP.forEach((key, value)-> {
                try {
                    List<Monitor> monitorTimeList = Collections.synchronizedList(new ArrayList<>());
                    // 遍历list
                    value.forEach(
                            monitor->{
                                if (monitor.getEndTime() > now) {
                                    monitorTimeList.add(monitor);
                                }
                            }
                    );

                    // 排序
                    monitorTimeList.sort(Comparator.comparing( Monitor::getTakeTime ));
                    //计算TP90的index
                    int size = monitorTimeList.size();
                    int tp90Index = (int) (size * 0.90f);
                    int tp99Index = (int) (size * 0.99f);

                    System.out.println(key + "====>>>> TP90耗时:" +
                            monitorTimeList.get( tp90Index ).getTakeTime() +
                            "ms;TP99耗时:" + monitorTimeList.get( tp99Index ).getTakeTime() + "ms");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        System.out.println();
//        }
    }

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        String methodName = invocation.getMethodName();

        // 开始时间
        long startTime = System.currentTimeMillis();
        Result invoke = invoker.invoke(invocation);
        // 结束时间
        long endTime = System.currentTimeMillis();
        // 计算耗时
        long takeTime = endTime - startTime;
        Monitor monitor = new Monitor(startTime, endTime, takeTime);

        // 存放到集合中
        if (MONITOR_MAP.containsKey(methodName)) {
            MONITOR_MAP.get(methodName).add(monitor);
        } else {
            List<Monitor> monitorList = Collections.synchronizedList(new ArrayList<>());
            monitorList.add(monitor);
            MONITOR_MAP.put(methodName, monitorList);
        }

        return invoke;
    }
}
